package com.wk.springboot.bean;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.wk.springboot.contants.Consts;
import com.wk.springboot.util.ESLog;
import com.wk.springboot.util.EsClientUtils;
import com.wk.springboot.util.EsUtils;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.tomcat.util.http.fileupload.IOUtils;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;

/**
 * @Classname EsImportBean
 * @Description 导入ES
 * @Date 2020/5/29 16:05
 * @Author by weikai
 */
@Configuration
public class EsImportBean implements InitializingBean {


  /**
   * 数据的_id
   */
  private static String esID;

  /**
   * 从JSON串中获取主键字段，用来作为ES文档的_id映射字段
   * 默认json串中应只有一个以id结尾的字段，若出现多个id，以最后一个id为准
   *
   * @param json
   * @return
   */
  private static void getID(String json) {
    LinkedHashMap<String, String> jsonMap = JSONObject
        .parseObject(json, new TypeReference<LinkedHashMap<String, String>>() {
        });
    for (String key : jsonMap.keySet()) {
      if (key.toLowerCase().endsWith("id")) {
        esID = key;
      }
    }
  }


  /**
   * 从JSON串中获取主键字段值,作为ES文档的_id
   *
   * @param json
   * @return
   */
  private static String getIDContent(String json, String jsonID) {
    JSONObject jsonObject = (JSONObject) JSON.parse(json);
    String id = jsonObject.getString(jsonID);
    Map<String, String> jsonMap = (Map) JSON.parse(json);
    return jsonMap.get(jsonID);
  }

  /**
   * 导入ES
   */
  private void importEs(Map<String,Object> importMap){
    StringBuffer command = new StringBuffer();
    // 如果配置target_table且不为空则使用target_table配置的index和type
    String index = (String) importMap.get("esDefaultIndex");
    String type = (String) importMap.get("esDefaultType");
    String target_table = (String) importMap.get("target_table");
    if (StringUtils.isNotEmpty(target_table)) {
      index = StringUtils.substringBefore(target_table, ".");
      type = StringUtils.substringAfter(target_table, ".");
    }
//    String ip = (String) importMap.get("ip");
    String source_table = (String)importMap.get("source_table");
//    int port = (Integer) importMap.get("port");
//    ESLog.LOG.info("数据导入ES开始,ip={};port={};index={};type={}", ip, port, index, type);
    TransportClient client = null;
    try (BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(new File(source_table)), "UTF-8"))) {
      // 从缓存Map中获取ES连接
//      client = EsUtils.getElasitcSearchClient(ip, (String) importMap.get("clusterName"), port,(String) importMap.get("user"));
      client = EsClientUtils.createClient();
      command.append("bulkInto " + index + ":" + type + " as " + source_table);
      //判断index在es中是否已存在，若不存在，则预创建一个空index
      if (!EsUtils.isExistsIndex(client, index)) {
        EsUtils.createEmptyIndex(client, index);
      }
      //是否清空原type中的数据
      if ((Boolean )importMap.get("isTruncate")) {
        EsUtils.cleanIndex(client, index, type);
      }
      //批量更新时将索引设置为不刷新，没有复本
      EsUtils.changeIndexProperties(index, client, Consts.PRE_REFRESH_INTERVAL, Consts.PRE_NUMBER_OF_REPLICAS);
      BulkRequestBuilder bulkRequest = client.prepareBulk();
      long begintime = System.currentTimeMillis();
      String jsonData = br.readLine();
      getID(jsonData);
      if (esID == null) {
        ESLog.LOG.info("未指明以ID作为结尾的字段作索引_ID,采用ES默认主键_ID");
      }
      // 批大小单位kb
      int es_bulk_size = (int) importMap.get("es_bulk_size");
      // 开启批量插入
      while (jsonData != null) {
        JSONObject jsonObject = (JSONObject) JSON.parse(jsonData);
        //修改批大小逻辑，将批大小由原来的数据条数修改为字节大小
        if (bulkRequest.request().estimatedSizeInBytes()> es_bulk_size) {
          begintime = System.currentTimeMillis();
          bulkRequest.execute().actionGet();
          //每次批量提交后，清空BulkRequestBuilder的缓存。否则存入数据会发生重复。
          //产生重复的两个条件是：1.没有清空BulkRequestBuilder； 2.没有指明每条记录的_id，重复数据不能以更新的形式避免重复
          bulkRequest.request().requests().clear();
          bulkRequest = null;
          bulkRequest = client.prepareBulk();
          //在导入数据时定期清理缓存
          EsUtils.clearIndicesCache(client, index);
          ESLog.LOG.info("本批次插入数据用时:{}", (System.currentTimeMillis() - begintime));
        }
        //如果json串中指定了_id，则
        if (StringUtils.isNotEmpty(esID)) {
          bulkRequest.add(client.prepareIndex(index, type, getIDContent(jsonData, esID)).setSource(jsonData, XContentType.JSON));
        } else {
          bulkRequest.add(client.prepareIndex(index, type).setSource(jsonData, XContentType.JSON));
        }
        jsonData = br.readLine();
      }
      IOUtils.closeQuietly(br);
      bulkRequest.execute().actionGet();
    } catch (Exception e) {
      ESLog.LOG.error(source_table + "导入出错", e);
    } finally {
      if (client != null) {
        EsUtils.changeIndexProperties(index, client, Consts.AFTER_REFRESH_INTERVAL, Consts.AFTER_NUMBER_OF_REPLICAS);
      }
      client.close();
    }

  }

  @Override
  public void afterPropertiesSet() throws Exception {

    //设置传入的参数
    Map<String,Object> importMap = new HashMap<>();
    importMap.put("esDefaultIndex","yl");
    importMap.put("esDefaultType","esyl");
    importMap.put("target_table","");
    importMap.put("source_table","S:\\工作目录\\es\\yltest.dat");
    importMap.put("isTruncate",false);
    importMap.put("es_bulk_size",10*1024);
    importEs(importMap);
  }
}
